package com.heima.behavior.service.impl;


import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.heima.behavior.mapper.ApArticleLikesMapper;
import com.heima.behavior.service.ApArticleLikesService;
import com.heima.common.constants.BehaviorConstants;
import com.heima.common.constants.BehaviorMessageConstants;
import com.heima.common.constants.HotArticleConstants;
import com.heima.common.redis.CacheService;
import com.heima.model.behavior.dtos.LikesBehaviorDto;
import com.heima.model.behavior.pojos.ApArticleLikes;
import com.heima.model.common.dtos.ResponseResult;
import com.heima.model.common.enums.AppHttpCodeEnum;
import com.heima.model.mess.UpdateArticleMess;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * 文章点赞行为服务层实现类
 *
 * @author huangjianfei
 * @date 2024/11/8
 */
@Service
@RequiredArgsConstructor
@Slf4j
public class ApArticleLikesServiceImpl extends ServiceImpl<ApArticleLikesMapper, ApArticleLikes> implements ApArticleLikesService {

    private final ApArticleLikesMapper apArticleLikesMapper;

    private final KafkaTemplate<String,String> kafkaTemplate;

    private final CacheService cacheService;


    /**
     * 文章点赞行为
     * @param dto
     * @param userId
     */
    @Override
    public ResponseResult likeAction(LikesBehaviorDto dto, Integer userId) {
        // 1、检查db是否有点赞记录
        ApArticleLikes apArticleLikes = apArticleLikesMapper.selectByArticleIdAndUserId(dto.getArticleId(), userId);
        ApArticleLikes likes = new ApArticleLikes();
        likes.setArticleId(dto.getArticleId());
        likes.setUserId(userId);
        likes.setLikeAction(dto.getOperation());
        likes.setLikeTimestamp(new Date());
        // 1、添加到数据库中
        if (apArticleLikes == null) {
            // 1.1、没有点赞记录，新增点赞记录
            if (Objects.equals(likes.getLikeAction(), BehaviorConstants.ARTICLE_LIKE_ACTION)) {
                apArticleLikesMapper.insertApArticleLikes(likes);
            } else {
                return ResponseResult.errorResult(AppHttpCodeEnum.PARAM_INVALID);
            }
        } else {
            // 1.2、更新点赞状态
            if (Objects.equals(likes.getLikeAction(), apArticleLikes.getLikeAction())) {
                // 点赞状态一致 直接返回
                return ResponseResult.okResult(AppHttpCodeEnum.SUCCESS);
            }
            // 点赞状态不一致，更新状态
            apArticleLikesMapper.updateApArticleLikes(likes);
        }
        // 1.3使用kafka stream流 更新文章点赞数
        asyncKafkaStreamUpdateLikesToDb(likes);
        // 1.3、修改文章点赞数 - kafka异步消息更新 文章点赞数 day09
        // asyncUpdateLikesToDb(likes);

        // 2、同步到Redis中
        // 2.1、更新点赞记录
        String likeRecordKey=BehaviorConstants.ARTICLE_LIKE + "_" + dto.getArticleId();
        // 2.1.1、设置过期时间
        cacheService.expire(likeRecordKey,60*30, TimeUnit.SECONDS);
        // 2.1.2、设置点赞行为
        Object o = cacheService.hGet(likeRecordKey, String.valueOf(userId));
        if(o!=null){
            Short aShort = Short.valueOf((String) o);
            if(!aShort.equals(likes.getLikeAction())){
                cacheService.hPut(likeRecordKey,String.valueOf(userId),String.valueOf(likes.getLikeAction()));
            }
        }else{
            cacheService.hPut(likeRecordKey,String.valueOf(userId),String.valueOf(likes.getLikeAction()));
        }

        // 2.2、更新文章点赞数 TODO:定时任务同步到Redis
        asyncUpdateLikesCountToRedis(likes);

        return ResponseResult.okResult(AppHttpCodeEnum.SUCCESS);
    }

    /**
     * 使用kafka stream数据聚合 更新 数据库文章的点赞数
     * @param likes
     * @Author huangjianfei
     * @Since 2024/11/15
     * @Version day11
     */
    private void asyncKafkaStreamUpdateLikesToDb(ApArticleLikes likes) {
        UpdateArticleMess mess=new UpdateArticleMess();
        mess.setArticleId(likes.getArticleId());
        mess.setType(UpdateArticleMess.UpdateArticleType.LIKES);

        if(Objects.equals(likes.getLikeAction(), BehaviorConstants.ARTICLE_LIKE_ACTION)){
            mess.setAdd(1);
        }else{
            mess.setAdd(-1);
        }
        // 发送消息 至 kafka 数据聚合
        kafkaTemplate.send(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC,JSON.toJSONString(mess));
    }

    /**
     * 异步更新Redis中的文章点赞数
     * @param likes 点赞文章的行为数据
     * @Author huangjianfei
     * @Since 2024/11/08
     * @Version day09
     */
    private void asyncUpdateLikesCountToRedis(ApArticleLikes likes) {
        log.info("kafka异步发送消息-更新点赞数到Redis：文章id{}", likes.getArticleId());
        Map<String, Object> map= new HashMap();
        map.put("articleId", likes.getArticleId());
        kafkaTemplate.send(BehaviorMessageConstants.APP_ARTICLE_LIKES_COUNT_TOPIC, JSON.toJSONString(map));
    }

    /**
     * 异步更新Db中的文章点赞数
     * @param likes 点赞文章的行为数据
     * @Author huangjianfei
     * @Since 2024/11/08
     * @Version day09
     */
    private void asyncUpdateLikesToDb(ApArticleLikes likes) {
        log.info("kafka异步发送消息-更新Db中的文章点赞数：文章id{},点赞操作{}", likes.getArticleId(), likes.getLikeAction());
        Map<String, Object> map = new HashMap();
        map.put("articleId", likes.getArticleId());
        map.put("likeAction", likes.getLikeAction());
        kafkaTemplate.send(BehaviorMessageConstants.APP_ARTICLE_LIKES_TOPIC, JSON.toJSONString(map));
    }
}
